import org.apache.spark.{SparkConf, SparkContext}

object IP2Addr {

  // 将ip地址转换为Long  ,便于查找目标IP的区间范围
  //>>>>>> ip2Long: (ip: String)Long
  def ip2Long(ip: String): Long = {
    ip.split("\\.")
      .map(_.toLong)
      .fold(0L) { (buffer, elem) =>
        buffer << 8 | elem
      }
  }

  // 给定ip地址，在ips中查找对应的城市名。使用二分查找算法
  //>>>>>> getCityName: (ip: Long, ips: Array[(Long, Long, String)])String
  def getCityName(ip: Long, ips: Array[(Long, Long, String)]): String = {
    var start = 0
    var end = ips.length - 1
    var middle = 0

    while (start <= end) {
      middle = (start + end) / 2
      if ((ip >= ips(middle)._1) && (ip <= ips(middle)._2))
        return ips(middle)._3
      else if (ip < ips(middle)._1)
        end = middle - 1
      else
        start = middle + 1
    }
    "Unknown"
  }

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val fileIP = "file:///F:\\lagou\\lagouhomework\\stage_4_module_2\\1、找到ip所属区域\\IP2Addr\\data\\ip.dat";
    val fileLog = "file:///F:\\lagou\\lagouhomework\\stage_4_module_2\\1、找到ip所属区域\\IP2Addr\\data\\http.log";
    // 读数据
    val iplines = sc.textFile(fileIP).collect()
        iplines.take(3).foreach(println)
    /**
     * 1.238.208.0|221.238.211.255|3723415552|3723416575|亚洲|中国|天津|天津|津南|电信|120112|China|CN|117.38537|38.99139
     * 221.238.212.0|221.238.212.255|3723416576|3723416831|亚洲|中国|天津|天津|滨海新区|电信|120116|China|CN|117.70162|39.02668
     * 221.238.213.0|221.238.213.255|3723416832|3723417087|亚洲|中国|天津|天津|南开|电信|120104|China|CN|117.15074|39.13821
     */

    //解析，收回来。最后变为广播变量
    val ipData = sc.textFile(fileIP).map { line =>
      val field = line.split("\\|")
      (field(2).toLong, field(3).toLong, field(6))
    }.collect()

    val loglines = sc.textFile(fileLog).collect()
        loglines.take(3).foreach(println)

    /**
     * 20090121001846493215000|115.120.8.171|414.kdd.cc|/s/i/?mn=WangYe&u=1&k=%CC%E1%C1%BA%D8%D5|Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; .NET CLR 2.0.50727)|http://414.kdd.cc/s/1/|searchk=%25CC%25E1%25C1%25BA%25D8%25D5; searchu=1; AJSTAT_ok_times=267; AJSTAT_ok_pages=4; ASPSESSIONIDASTRARRC=FLNNBNJCLNBLFLPBPBJPDBFE
     */


    //将ip地址转换为Long类型
    val httpData = sc.textFile(fileLog).map(x => ip2Long(x.split("\\|")(1)))
    httpData.collect().take(3).foreach(println)
    /**
     * 2111136891
     * 1969608581
     * 1969610308
     */

    val ipBC = sc.broadcast(ipData.sortBy(_._1))

    httpData.mapPartitions { iter =>
      val ipsInfo: Array[(Long, Long, String)] = ipBC.value
      iter.map { ip =>
        val city: String = getCityName(ip, ipsInfo)
        (city, 1)
      }
    }.reduceByKey(_ + _).collect().foreach(println)

    /**
     * (陕西,1824)
     * (河北,383)
     * (云南,126)
     * (重庆,868)
     */

    //拿到ip地址和该行日志的对应关系，方便后面 替换ip为城市名
    val ip2lineData = sc.textFile(fileLog).map(x => (ip2Long(x.split("\\|")(1)), x.split("\\|")(1), x))
    ip2lineData.collect().take(3).foreach(println)
    //匹配，替换
    ip2lineData.mapPartitions { iter =>
      val ipsInfo: Array[(Long, Long, String)] = ipBC.value
      iter.map { case (ipValue, ipAddr, line) =>
        val city: String = getCityName(ipValue, ipsInfo)
        val result = line.replace(ipAddr, city);
        result
      }
    }.collect().take(3).foreach(println)
    // 关闭SparkContext
    sc.stop()
  }
}
